//! struce KvStore
//! 有着未知的bug
//! 不知道为什么test单独运行总是成功， 一起运行有时会失败一两个。。。

use std::collections::{HashMap, HashSet};
use std::fs::{File, OpenOptions};
use std::io::prelude::*;
use std::io::SeekFrom;
use std::mem::size_of;
use std::ops::Drop;
use std::path::PathBuf;

use crate::{KvsError::*, Result};

use serde::{Deserialize, Serialize};

#[derive(Deserialize, Serialize)]
enum Cmd {
    Set { key: String, value: String },
    Rm { key: String },
}

pub struct KvStore {
    /// key and log pointer
    index: HashMap<String, Index>,
    buffer: HashMap<String, String>,
    deleted: HashSet<String>,
    /// 现在正在继续写入的 文件 id
    active_file: u32,
    /// 当前目录, 也是数据文件存放的目录
    path: PathBuf,
    file_end_pos: usize,
    // old_files: Vec<PathBuf>,
    // index_file: PathBuf,
    // buffer_count_down = 0时, 将buffer写入硬盘
    buffer_counter: Counter,
    compacte_counter: Counter,
}

impl KvStore {
    /// 读取目录下的特定文件重建索引
    pub fn open(path: impl Into<PathBuf>) -> Result<Self> {
        // 思路是 一开始 读配置文件, 然后确定 active_file
        // 是否只在退出的时候修改 index_file
        let path: PathBuf = path.into();
        let active_file = 1;
        std::fs::create_dir_all(path.clone())?;
        std::env::set_current_dir(&path)?;
        let mut file = OpenOptions::new()
            .write(true)
            .read(true)
            .create(true)
            .open(file_id_to_name(active_file))?; // create it if not exist
                                                  // let mut file = File::open(file_id_to_name(active_file))?;
        let file_end_pos = file.metadata()?.len() as usize;
        let mut index = HashMap::new();
        let mut file_offset: usize = 0;
        while file_offset < file_end_pos {
            // 毫无疑问, 这样会重复读取多次
            let (key, value) = match read_index_from_disk(&mut file, &mut file_offset) {
                Ok(kv) => kv,
                Err(err) => {
                    return Err(err);
                }
            };
            if value.value_version > 0 {
                index.insert(key, value);
            } else {
                if index.contains_key(&key) {
                    index.remove(&key);
                }
            }
        }
        Ok(KvStore {
            index,
            path,
            active_file,
            file_end_pos,
            buffer: HashMap::new(),
            deleted: HashSet::new(),
            compacte_counter: Counter::new(100),
            buffer_counter: Counter::new(100),
        })
    }

    /// 根据索引获得值
    pub fn get(&self, key: String) -> Result<Option<String>> {
        // 说实话我不太理解这里为什么要多套一层 option
        if self.deleted.contains(&key) {
            // 即使 deleted 与 buffer 不重合, 但是它与 index 可能重合, 所以依旧需要检查
            return Ok(None);
        }
        if self.buffer.contains_key(&key) {
            return Ok(self.buffer.get(&key).cloned());
        }
        self.get_from_disk(key)
    }

    /// 更新对应索引值， 没有则创建它
    pub fn set(&mut self, key: String, value: String) -> Result<()> {
        if self.deleted.contains(&key) {
            self.deleted.remove(&key);
        }
        self.buffer.insert(key, value);
        self.count_down()?;
        Ok(())
    }

    /// 删除索引及值
    pub fn remove(&mut self, key: String) -> Result<()> {
        if self.buffer.contains_key(&key) {
            // 保证 buffer 和 deleted 不重合
            self.buffer.remove(&key);
        } else {
            if self.deleted.contains(&key) || !self.index.contains_key(&key) {
                return Err(NoKey { key });
            }
            self.deleted.insert(key);
        }
        self.count_down()?;
        Ok(())
    }

    /// 修改会优先存储到 buffer， 但是没过 size 次， 会将buffer写入硬盘
    pub fn set_buffer_size(&mut self, size: u64) {
        self.buffer_counter.set(size);
    }

    /// 每修改多少次， 压缩一次
    pub fn set_compact_count_down(&mut self, count: u64) {
        self.compacte_counter.set(count);
    }

    fn index_contains_key(&self, key: &str) -> bool {
        let in_idx = self.index.get(key);
        if in_idx.is_none() || in_idx.unwrap().value_version < 0 {
            false
        } else {
            true
        }
    }

    fn get_from_disk(&self, key: String) -> Result<Option<String>> {
        if self.index_contains_key(&key) {
            let idx = self.index.get(&key).unwrap();
            if idx.value_version < 0 {
                return Ok(None);
            }
            let mut file = File::open(file_id_to_name(self.active_file))?;
            // 懒得读 version
            file.seek(SeekFrom::Start((idx.value_pos + size_of::<i32>()) as u64))?;
            let mut key_size_buf = [0; size_of::<usize>()];
            let mut value_size_buf = [0; size_of::<usize>()];
            file.read_exact(&mut key_size_buf)?;
            file.read_exact(&mut value_size_buf)?;
            let key_size = usize::from_ne_bytes(key_size_buf);
            let value_size = usize::from_ne_bytes(value_size_buf);
            file.seek(SeekFrom::Current(key_size as i64))?;
            let mut value_buf = vec![0; value_size];
            file.read_exact(&mut value_buf)?;
            return Ok(Some(String::from_utf8(value_buf).unwrap())); // TODO
        }
        Ok(None)
    }

    fn with_index() -> Result<Self> {
        todo!()
    }

    fn count_down(&mut self) -> Result<()> {
        if self.buffer_counter.count_down() {
            self.buffer_to_disk()?;
        }
        if self.compacte_counter.count_down() {
            self.compacte()?;
        }
        Ok(())
    }

    fn rewrite_index_file(&mut self) {
        todo!()
    }

    fn buffer_to_disk(&mut self) -> Result<()> {
        let mut file = OpenOptions::new()
            .append(true)
            .open(file_id_to_name(self.active_file))?;
        for (key, value) in self.buffer.iter() {
            let mut version = 1; // 从 0 开始的话, 0 取负可能会出现一些 bug
            if let Some(idx) = self.index.get(key) {
                version = idx.value_version.abs() + 1;
            }
            let size = write_to_disk(&mut file, &mut self.file_end_pos, key, value, version)?; // 为什么不借用 self 呢, 因为不让...
            self.index.insert(
                // 这块其实可以放进 write 的, TODO
                key.clone(),
                Index {
                    value_version: version,
                    value_pos: self.file_end_pos - size,
                    size: size as u32,
                },
            );
        }

        for key in self.deleted.iter() {
            if self.index.contains_key(key) {
                let version = -self.index.get(key).unwrap().value_version;
                let size = write_to_disk(&mut file, &mut self.file_end_pos, key, "", version)?;
                self.index.insert(
                    key.clone(),
                    Index {
                        value_version: version,
                        value_pos: self.file_end_pos - size,
                        size: size as u32,
                    },
                );
            }
        }
        self.deleted = HashSet::new();
        self.buffer = HashMap::new();
        Ok(())
    }

    /// 压缩
    pub fn compacte(&mut self) -> Result<()> {
        // self.buffer_to_disk()?; // 其实不是必须
        let tmp_file_name = format!("tmp_data{}", self.active_file);
        let mut new_file = File::create(&tmp_file_name)?;
        let mut new_index: HashMap<String, Index> = HashMap::new();
        let mut new_offset = 0;
        for (k, v) in self.index.iter() {
            if let Ok(Some(value)) = self.get_from_disk(k.clone()) {
                let size = write_to_disk(&mut new_file, &mut new_offset, k, &value, 1)?;
                new_index.insert(
                    k.clone(),
                    Index {
                        value_pos: new_offset - size,
                        value_version: 1,
                        size: (new_offset - size) as u32,
                    },
                );
            }
        }
        self.index = new_index;
        self.file_end_pos = new_offset;
        std::fs::rename(tmp_file_name, file_id_to_name(self.active_file))?;
        Ok(())
    }
}

impl Drop for KvStore {
    fn drop(&mut self) {
        if let Err(err) = self.buffer_to_disk() {
            eprintln!("can not write buffer to file");
        }; // 这里如果出错该如何解决我真不知道...
    }
}

#[derive(Debug)]
struct Index {
    // hash: String,
    // file_id: PathBuf,
    value_pos: usize,
    value_version: i32, // 负值表示被删除
    size: u32,          // 包括key, value, key_size, value_size
}

fn file_id_to_name(file_id: u32) -> String {
    format!("data{}.kv", file_id.to_string())
}

fn read_index_from_disk(file: &mut File, file_offset: &mut usize) -> Result<(String, Index)> {
    file.seek(SeekFrom::Start(*file_offset as u64))?;
    let mut version_buf = [0; size_of::<i32>()];
    file.read_exact(&mut version_buf)?;
    let version = i32::from_ne_bytes(version_buf);
    let mut key_size_buf = [0; size_of::<usize>()];
    let mut value_size_buf = [0; size_of::<usize>()];
    file.read_exact(&mut key_size_buf)?;
    file.read_exact(&mut value_size_buf)?;
    let key_size = usize::from_ne_bytes(key_size_buf);
    let value_size = usize::from_ne_bytes(value_size_buf);
    let size = key_size + value_size + size_of::<usize>() * 2 + size_of::<i32>();
    *file_offset += size; // 不管 有没有被删除, 读的时候都必须修改指针
    let mut key_buf = vec![0; key_size];
    file.read_exact(&mut key_buf)?;
    let key: String = String::from_utf8(key_buf).unwrap(); // TODO: 吧 FromUtf8Error 加入 KvsError
    Ok((
        key,
        Index {
            size: size as u32,
            value_version: version,
            value_pos: *file_offset - size,
        },
    ))
}

fn write_to_disk(
    // 莫名感觉这玩意不该提到 impl 外面来
    file: &mut File,
    file_offset: &mut usize,
    key: &str,
    value: &str,
    version: i32,
) -> Result<usize> {
    // 在硬盘里 key-value pair 是这样的
    // version<i32> key_size<usize> value_size<usize> key<String> value<String>
    file.seek(SeekFrom::Start(*file_offset as u64))?;
    file.write_all(&version.to_ne_bytes())?; //  i32
    file.write_all(&key.len().to_ne_bytes())?; // usize
    file.write_all(&value.len().to_ne_bytes())?; // usize
    file.write_all(key.as_bytes())?; // key.len()
    file.write_all(value.as_bytes())?; // value.len()
    let size = size_of::<usize>() * 2 + size_of::<i32>() + key.len() + value.len();
    *file_offset += size;
    Ok(size)
}

struct Counter {
    current: u64,
    start: u64,
}

impl Counter {
    fn new(start: u64) -> Self {
        Counter {
            current: start,
            start,
        }
    }

    fn count_down(&mut self) -> bool {
        self.current -= 1;
        if self.current != 0 {
            return false;
        }
        self.current = self.start;
        true
    }

    fn set(&mut self, start: u64) {
        self.start = start;
    }
}
